package com.besttone.util;

import com.besttone.bean.BusinessTypeByDayBean;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by zhengzhenzhen on 2016/11/22.
 */
public class HbaseClientUtil {
    private static Log _log = LogFactory.getLog(HbaseClientUtil.class);
    private static Connection connection;

    /**
     * 根据rowkey集合查询SmsDay记录
     * @param rowKeys
     */
    public static List<BusinessTypeByDayBean> querySmsDayByRowKeys(List<String> rowKeys){
        List<BusinessTypeByDayBean> smsDayList=new ArrayList<BusinessTypeByDayBean>();
        Connection connection =getHbaseConnection();
        try{
        for(String rowKey:rowKeys){
            Table table = connection.getTable(TableName.valueOf("tb_sms_day"));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result=table.get(get);
            Map<String,Cell> map= HbaseObject.getMapFromCell(result.listCells());
            BusinessTypeByDayBean smsDayBean=(BusinessTypeByDayBean)HbaseObject.toBean(map,BusinessTypeByDayBean.class);
            smsDayList.add(smsDayBean);
        }
    } catch (IOException e) {
            e.printStackTrace();
            _log.info("hbase查询smsday异常:"+e);
    }
        return smsDayList;
    }
    /**
     * 获取Hbase connection
     *
     * @return
     */
    private static synchronized Connection getHbaseConnection() {
        if (connection == null) {
            try {
                _log.info("创建Connection");
                Configuration conf = HBaseConfiguration.create();
                conf.set("hbase.zookeeper.quorum", PropertieUtils.getProperty("hbase.zookeeper.quorum"));
                conf.set("hbase.zookeeper.property.clientPort", PropertieUtils.getProperty("hbase.zookeeper.property.clientPort"));
                conf.set("hadoop.home.dir",PropertieUtils.getProperty("hadoop.home.dir"));
                connection = ConnectionFactory.createConnection(conf);
                return connection;
            } catch (Exception e) {
                _log.info("connection创建异常:" + e);
            }
        }
        return connection;
    }
}
